package org.damocode.iot.network.udp.gateway;

import lombok.extern.slf4j.Slf4j;
import org.damocode.iot.core.device.DeviceOperator;
import org.damocode.iot.core.device.DeviceOperatorManager;
import org.damocode.iot.core.message.DeviceMessage;
import org.damocode.iot.core.message.Message;
import org.damocode.iot.core.message.codec.*;
import org.damocode.iot.core.protocol.ProtocolSupport;
import org.damocode.iot.core.server.DecodedClientMessageHandler;
import org.damocode.iot.core.server.session.DeviceSession;
import org.damocode.iot.core.server.session.DeviceSessionManager;
import org.damocode.iot.network.udp.UdpMessage;
import org.damocode.iot.network.udp.UdpSupport;
import org.damocode.iot.network.utils.DeviceGatewayHelper;
import rx.Subscription;
import rx.functions.Action1;
import rx.subjects.PublishSubject;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * @Description:
 * @Author: zzg
 * @Date: 2022/7/19 15:58
 * @Version: 1.0.0
 */

@Slf4j
public class UdpDeviceGateway {

    private final UdpSupport udpSupport;

    private final ProtocolSupport protocolSupport;

    private final DeviceOperatorManager deviceOperatorManager;

    private final DeviceGatewayHelper helper;

    PublishSubject<Message> subject = PublishSubject.create();

    private final AtomicBoolean started = new AtomicBoolean(false);

    private Subscription subscription;

    public UdpDeviceGateway(UdpSupport udpSupport, DeviceSessionManager sessionManager, ProtocolSupport protocolSupport, DeviceOperatorManager deviceOperatorManager, DecodedClientMessageHandler messageHandler) {
        this.udpSupport = udpSupport;
        this.protocolSupport = protocolSupport;
        this.deviceOperatorManager = deviceOperatorManager;
        this.helper = new DeviceGatewayHelper(deviceOperatorManager,sessionManager,messageHandler);
    }

    public Transport getTransport() {
        return DefaultTransport.UDP;
    }

    public PublishSubject<Message> onMessage() {
        return subject;
    }

    public void startup() {
        doStart();
    }

    public void shutdown() {
        started.set(false);
        this.subject.onCompleted();
        if (this.subscription != null) {
            this.subscription.unsubscribe();
        }
    }

    private void doStart() {
        if (started.getAndSet(true) && subscription != null) {
            return;
        }

        subscription = udpSupport.subscribe().subscribe(message -> {
            log.debug("收到UDP[{}]报文:\n{}", message.getAddress(), message);
            if(started.get()){
                handleUdpMessage(message);
            }
        });
    }

    void handleUdpMessage(UdpMessage message){
        DeviceMessageCodec codec = protocolSupport.getMessageCodec(getTransport());
        if(codec == null){
            return;
        }
        AtomicReference<Duration> keepAliveTimeoutRef = new AtomicReference<>();
        DeviceMessage deviceMessage = null;
        try {
            deviceMessage = codec.decode(new FromDeviceMessageContext() {
                @Override
                public DeviceSession getSession() {
                    return new UnknownUdpDeviceSession(udpSupport, message.getAddress(), getTransport(), protocolSupport) {
                        @Override
                        public void setKeepAliveTimeout(Duration timeout) {
                            keepAliveTimeoutRef.set(timeout);
                        }
                    };
                }

                @Override
                public EncodedMessage getMessage() {
                    return message;
                }

                @Override
                public DeviceOperator getDevice(String deviceId) {
                    return deviceOperatorManager.getDevice(deviceId);
                }
            });
        }catch (Exception e) {
            if(log.isDebugEnabled()){
                e.printStackTrace();
            }
        }
        if(deviceMessage != null){
            handleDeviceMessage(deviceMessage,message,keepAliveTimeoutRef);
        }
    }

    void handleDeviceMessage(DeviceMessage msg, UdpMessage message,AtomicReference<Duration> keepAliveTimeoutRef){
        if(subject.hasObservers()){
            subject.onNext(msg);
        }
        Function<DeviceOperator, DeviceSession> sessionBuilder = device -> new UdpDeviceSession(device, udpSupport, message.getAddress(), getTransport(),protocolSupport);
        Consumer<DeviceSession> sessionConsumer = DeviceGatewayHelper.applySessionKeepaliveTimeout(msg,keepAliveTimeoutRef::get);
        sessionConsumer.andThen(session -> {
            UdpDeviceSession deviceSession = session.unwrap(UdpDeviceSession.class);
            deviceSession.setAddress(message.getAddress());
            deviceSession.setSupport(udpSupport);
        });
        helper.handleDeviceMessage(msg,sessionBuilder,sessionConsumer);
    }

}
